package com.htdata.iiot.api.tsdb.utils;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Timer;
import com.htdata.iiot.api.tsdb.config.Configs;
import com.htdata.iiot.api.tsdb.pojo.DataPoint;
import com.htdata.iiot.api.tsdb.pojo.DeviceNumber;
import com.htdata.iiot.api.tsdb.pojo.Metric;
import com.htiiot.common.config.ConfigFactory;
import net.opentsdb.core.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.LongAdder;

/**
 * Created by SunMy on 2017/7/6.
 */

public class QueryTSDB {
    private static final Logger logger = LogManager.getLogger(QueryTSDB.class);
    private String tsdbConfName = Configs.TSDBConfig.CONFIG_SECTION;
    private TSDB tsdb;
    private Timer tsdbTimer = new Timer();
    private Counter tsdbDelteCounter = new Counter();
    private Counter tsdbDelteErrorCounter = new Counter();

    public QueryTSDB() {
        this.initParamters();
    }


    /**
     * init parameters
     */
    private void initParamters() {
        logger.info("initing TSDB parameters");

        if (tsdb == null) {

            tsdb = new TSDBBuilder(ConfigFactory.getConfigMgr(), tsdbConfName).createTSDB();
        }

    }

    /**
     * 根据metricName,startTime,endTime,downsample获取数据
     *
     * @param metricName 测点名称
     * @param startTime  开始时间
     * @param endTime    结束时间
     * @param downsample 聚类算法,比如 3s-avg-zero
     * @return
     * @throws Exception
     */
    public DataPoint[] getTSData(String metricName, long startTime, long endTime, String downsample) throws Exception {

        return getTSData(metricName, new Date(startTime), new Date(endTime), downsample);
    }

    /**
     * 根据metricName,startTime,endTime,downsample获取数据
     *
     * @param metricName 测点名称
     * @param startTime  开始时间
     * @param endTime    结束时间
     * @param downsample 聚类算法,比如 3s-avg-zero
     * @return
     * @throws Exception
     */
    public DataPoint[] getTSData(String metricName, Date startTime, Date endTime, String downsample) throws Exception {
        Query[] tsdbQueries = null;
        try {
            TSQuery query = new TSQueryBuilder().setStartTime(startTime).setEndTime(endTime)
                    .setMetricName(metricName).setDownsample(downsample).build();
            query.validateAndSetQuery();
            tsdbQueries = query.buildQueries(tsdb);
        } catch (Exception e) {
            logger.error(e.getMessage());
            throw e;
        }

        return getDataFromQuery(tsdbQueries, startTime, endTime);
    }

    /**
     * get result from tsdb query
     *
     * @param tsdbqueries array list of {@link Query}
     * @return {@link DataPoint} arrays
     */
    private DataPoint[] getDataFromQuery(Query[] tsdbqueries, Date startTime, Date endTime) {
        ArrayList<DataPoints> result = new ArrayList<DataPoints>();
        if (tsdbqueries != null) {
            int nqueries = tsdbqueries.length;
            for (int i = 0; i < nqueries; i++) {
                try {
                    DataPoints[] dps = tsdbqueries[i].run();
                    result.addAll(Arrays.asList(dps));
                } catch (Exception ex) {
                    logger.error("error retrive data ", ex);
                }
            }
        }
        List<DataPoint> dpResult = new ArrayList<DataPoint>();
        long startTimeL = startTime.getTime();
        long endTimeL = endTime.getTime();
        for (DataPoints dps : result) {
            SeekableView it = dps.iterator();
            while (it.hasNext()) {

                DataPoint datapoint = new DataPoint();
                datapoint.setTags(dps.getTags());
                /*		Map<String, String> tags = dps.getTags();
				String tenantId = tags.get("tenantid");
				String deviceId = tags.get("deviceid");
				String metricId = tags.get("metricid");*/
                String metricName = dps.metricName();
                String dn = metricName.split("_")[0];


                DeviceNumber deviceNumber = DeviceNumber.fromHexString(dn);
                datapoint.setDeviceNumber(deviceNumber);
                Metric metric = new Metric();
                metric.setName(metricName);
                net.opentsdb.core.DataPoint dp = it.next();
                long ts = dp.timestamp();

                if (ts >= startTimeL && ts < endTimeL) {
                    metric.setValue(dp.doubleValue());
                    datapoint.setTs(dp.timestamp());
                    datapoint.setMetric(metric);
                    dpResult.add(datapoint);
                }

            }

        }

        return dpResult.toArray(new DataPoint[0]);
    }

    public boolean deleteTsdb(String metricName, long startime, long endtime) {
        Timer.Context context = this.tsdbTimer.time();
        try {

            TSQuery query = new TSQueryBuilder().setStartTime(new Date(startime)).setEndTime(new Date(endtime))
                    .setMetricName(metricName)
                    .build();
            query.validateAndSetQuery();
            Query[] tsdbQueries = query.buildQueries(tsdb);
            final int nqueries = tsdbQueries.length;
            for (int i = 0; i < nqueries; i++) {
                try {
                    tsdbQueries[i].setDelete(true);
                    tsdbQueries[i].run();
                    this.tsdbDelteCounter.inc();
                } catch (Exception ex) {
                    this.tsdbDelteErrorCounter.inc();
                    logger.error("error retrive data ", ex);
                    throw ex;
                }
            }
            return true;

        } catch (Exception e) {
            logger.error(e.getMessage());
            this.tsdbDelteErrorCounter.inc();
            throw e;
        } finally {
            context.close();
        }
    }


    public static LongAdder deleteAdder = new LongAdder();
    public static LongAdder errorAdder = new LongAdder();

    /**
     * 删除指定租户在某时间段的测点
     *
     * @param tid       租户id(多个请以逗号隔开,null为所有租户)
     * @param startT    开始时间
     * @param enT       截止时间(可以为空，空为到现在)
     * @param threadNum 线程数量
     */
    public void delete(String tid, long startT, long enT, int threadNum) throws InterruptedException {
        long start = System.currentTimeMillis();
        DBQuery dbQ = new DBQuery();
        List<String> metricL = dbQ.metircQuery(tid);
        int metricSize = metricL.size();
        CountDownLatch cd = new CountDownLatch(threadNum);
        if (!metricL.isEmpty()) {
            List<String>[] smallL = Tools.avgList(metricL, threadNum);
            for (int i = 0; i < smallL.length; i++) {
                List<String> metrics = smallL[i];
                new Thread(new MultiDelete(cd, metrics, startT, enT)).start();
            }
        }
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(5000l);
                        System.out.println("总计" + metricSize + "条; 已经删除:" + deleteAdder.sum() + "条; 删除失败:" + errorAdder.sum() + "条; 已完成:" + (deleteAdder.sum() + errorAdder.sum()) * 100 / metricSize + "%");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
        cd.await();
        logger.info("总计:成功删除:" + deleteAdder.sum() + "条;失败:" + errorAdder.sum() + "条;总耗时:" + (System.currentTimeMillis() - start) + "ms;");
    }
}

class MultiDelete implements Runnable {
    private org.apache.log4j.Logger log = org.apache.log4j.Logger.getLogger(MultiDelete.class);
    private CountDownLatch cd;
    private List<String> metrics;
    private long startT;
    private long enT;
    private QueryTSDB query = new QueryTSDB();

    public MultiDelete(CountDownLatch cd, List<String> metrics, long startT, long enT) {
        this.cd = cd;
        this.metrics = metrics;
        this.startT = startT;
        this.enT = enT;
    }

    @Override
    public void run() {
        for (String metric : metrics) {
            try {
                query.deleteTsdb(metric, startT, enT);
                System.out.println("metric:" + metric + ";already deleted!!!");
                QueryTSDB.deleteAdder.increment();
            } catch (Exception e) {
                log.error("metric:" + metric + ";delete wrong!!!\r\n" + e);
                QueryTSDB.errorAdder.increment();
            }
        }
        cd.countDown();
    }
}